package com.hdu.dwh.controller;

import com.hdu.dwh.dtos.cdc_dtos.*;
import org.apache.commons.beanutils.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.hdu.dwh.dao.FlinkBaseDao;
import com.hdu.dwh.dao.FlinkTableDao;
import com.hdu.dwh.dao.PaimonBaseDao;
import com.hdu.dwh.dtos.MysqlcdcWithSourceAccessIdDtos;
import com.hdu.dwh.enums.HttpStatusEnum;
import com.hdu.dwh.exception.CustomException;
import com.hdu.dwh.exception.FlinkWebException;
import com.hdu.dwh.pojos.FlinkJob;
import com.hdu.dwh.pojos.FlinkJobRequest;
import com.hdu.dwh.pojos.source_access_pojos.SourceAccess;
import com.hdu.dwh.pojos.source_access_pojos.TableRelation;
import com.hdu.dwh.pojos.source_access_pojos.DataWarehouseDatabase;
import com.hdu.dwh.pojos.source_access_pojos.DataWarehouseTable;
import com.hdu.dwh.results.ResponseResult;
import com.hdu.dwh.service.ISourceAccessService;
import com.hdu.dwh.service.TableRelationService;
import com.hdu.dwh.service.flink_service.FlinkJarService;
import com.hdu.dwh.service.flink_service.FlinkWebRestService;
import com.hdu.dwh.utils.CDCUtils;
import com.hdu.dwh.service.DataWarehouseDatabaseService;
import com.hdu.dwh.service.DataWarehouseTableService;
import com.hdu.dwh.vos.DwhTableCDCVo;

import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.extern.slf4j.Slf4j;

import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;


@RestController
@RequestMapping("/paimon_cdc_job")
@Slf4j
@Tag(name="CDC任务", description = "CDC任务接口")
public class CDCController {
    
    @Autowired
    private FlinkWebRestService flinkWebRestService;

    @Autowired
    private ISourceAccessService sourceAccessService;

    @Autowired
    private DataWarehouseTableService dwhTableService;

    @Autowired
    private DataWarehouseDatabaseService dwhDatabaseService;

    @Autowired
    private PaimonBaseDao paimonBaseDao;

    @Autowired
    private FlinkTableDao flinkTableDao;

    @Autowired
    private TableRelationService sourceTargetRelationService;

    @Autowired
    private FlinkJarService flinkJarService;

    @Value("${dwh.minio.endpoint}")
    private String endpoint;

    @Value("${dwh.minio.accessKey}")
    private String accessKey;

    @Value("${dwh.minio.secretKey}")
    private String secretKey;

    @Value("${dwh.flink.warehouse}")
    private String warehouse;

    @Value("${dwh.flink.jars.cdc-jar.jar-name}")
    private String cdcJarName;

    @Value("${dwh.flink.jars.sql-jar.jar-name}")
    private String radarJarName;

    @Operation(summary ="提交Mysql CDC任务")
    @PostMapping("mysqlCDC")
    public ResponseResult<FlinkJob> mysqlCDC(@RequestBody FlinkJobRequest<MysqlcdcDtos> flinkJobRequest) throws Exception {
        String jarId = flinkJarService.getJarIdbyName(this.cdcJarName);
//        CDCUtils.warehouse = this.warehouse;
        flinkJobRequest.setProgramArgsList(flinkJobRequest.getRequest().getMysqlCDCConfigList());

        if(flinkJobRequest.getProgramArgsList() == null){
            throw new FlinkWebException("Flink参数为空,无法提交任务");
        }
        try{
            flinkWebRestService.planJob(jarId, flinkJobRequest);
        } catch(Exception e){
            throw new FlinkWebException("Flink任务提交失败，请检查参数配置是否正确，错误信息", e.getMessage());
        }

        return ResponseResult.success("提交任务成功", flinkWebRestService.runJob(jarId, flinkJobRequest));
    }

    /**
     * 使用SourceAccessId和具体参数执行Mysql CDC任务
     * @param jarId
     * @param flinkJobRequest {@link FlinkJobRequest}<{@link MysqlcdcWithSourceAccessIdDtos}>
     * @return {@link ResponseResult}<{@link DwhTableCDCVo}>
     * @throws Exception
     */
    @PostMapping("MysqlCdcWithSourceAccessId")
    @Transactional
    @Operation(summary = "提交MysqlCDC携带SourceAccessId和具体参数")
    public synchronized ResponseResult<DwhTableCDCVo> MysqlCdcWithSourceAccessId(@RequestBody FlinkJobRequest<MysqlcdcWithSourceAccessIdDtos> flinkJobRequest) throws Exception {
        // TODO: 重新设计,将下列代码封装到service中
        Long sourceAccessId = flinkJobRequest.getRequest().getSourceAccessId();
        String jarId = flinkJarService.getJarIdbyName(this.cdcJarName);
        SourceAccess sourceAccess = sourceAccessService.getById(sourceAccessId);
        if(sourceAccess == null){
            return ResponseResult.fail("sourceAccessId无效，未找到对应的SourceAccess");
        }
        String sourceDatabase = flinkJobRequest.getRequest().getSourceDatabase();
        String sourceTable = flinkJobRequest.getRequest().getSourceTable();

        String targetDatabase = flinkJobRequest.getRequest().getTargetDatabase();
        String targetTable = flinkJobRequest.getRequest().getTargetTable();

        
        // 创建sourceTable和sourceDatabase
        DataWarehouseDatabase sourceDb = dwhDatabaseService.queryByNameAndSourceAccessIdAndSourceType(sourceDatabase, sourceAccess.getSourceType(), sourceAccessId);
        if (sourceDb == null) {
            sourceDb = dwhDatabaseService.createSourceDatabase(sourceAccessId, sourceDatabase, sourceAccess.getSourceType(), sourceAccess.getDataType());
        }
        
        DataWarehouseTable sourceTb = dwhTableService.queryByNameAndDatabaseId(null, sourceTable, sourceDb.getId());
        if (sourceTb == null && sourceDb != null) {
            // sourceTable的sourceAccessId为空
            sourceTb = dwhTableService.createTable(sourceDb.getId(), sourceTable, null, -1, "CDC同步的源表");
        }
        
        // 创建targetTable和targetDatabase
        DataWarehouseDatabase targetDb = dwhDatabaseService.queryByNameAndSourceAccessIdAndSourceType(targetDatabase, "paimon", null);
        if (targetDb == null) {
            if(!this.paimonBaseDao.checkDatabaseExists(targetDatabase)) {
                this.paimonBaseDao.createDatabase(targetDatabase);
            }
            targetDb = dwhDatabaseService.createSourceDatabase(null, targetDatabase, "paimon", 0);
        }
        
        DataWarehouseTable targetTb = dwhTableService.queryByNameAndDatabaseId(sourceTable, targetTable, targetDb.getId());
        if (targetTb == null && targetDb != null && sourceTb != null) {
            targetTb = dwhTableService.createTable(targetDb.getId(), targetTable, sourceTb.getId(), 0, "ODS层");
        }
        
        MysqlcdcDtos mysqlcdcDtos = new MysqlcdcDtos(
            sourceAccess.getHost(), 
            sourceAccess.getPort(), 
            sourceAccess.getUsername(), 
            sourceAccess.getPassword(),
            sourceDatabase,
            sourceTable,
            targetDatabase,
            targetTable
        );

        flinkJobRequest.setProgramArgsList(mysqlcdcDtos.getMysqlCDCConfigList());
        FlinkJobRequest<MysqlcdcDtos> mysqlRequest = new FlinkJobRequest<>();
        BeanUtils.copyProperties(mysqlRequest, flinkJobRequest);
        try{
            flinkWebRestService.planJob(jarId, mysqlRequest);
        } catch(Exception e) {
            throw new FlinkWebException("Flink任务提交失败，请检查参数配置是否正确，错误信息", e.getMessage());
        }
        FlinkJob job = flinkWebRestService.runJob(jarId, mysqlRequest);
        DwhTableCDCVo sourceTableVo = new DwhTableCDCVo(sourceAccess, sourceDb, sourceTb, targetDb, targetTb, job);
        if(sourceTb == null || sourceTb.getId() == null || targetTb == null || targetTb.getId() == null){
            throw new FlinkWebException("SourceTable或TargetTable未创建成功，无法建立关联");
        }else {
            this.sourceTargetRelationService.saveWithSourceTableIdAndTargetTableId(sourceTb.getId(), targetTb.getId());
        }
        // this.sourceTargetRelationService.save(new TableRelation(sourceAccessId, sourceTable, sourceAccessId, targetTable));
        return ResponseResult.success(sourceTableVo);
    }
    
    // TODO: 服务器还需配置相关信息
    @Deprecated
    @Operation(summary ="提交PostgreSQL CDC任务")
    @PostMapping("postgresqlCDC")
    public ResponseResult<FlinkJob> postgresqlCDC(@RequestBody FlinkJobRequest<PostgresqlcdcDtos> flinkJobRequest) throws Exception {
        String jarId = flinkJarService.getJarIdbyName(this.cdcJarName);
        flinkJobRequest.setProgramArgsList(flinkJobRequest.getRequest().getPostgresqlCDCConfigList());

        if(flinkJobRequest.getProgramArgsList() == null){
            throw new FlinkWebException("Flink参数为空,无法提交任务");
        }
        try{
            flinkWebRestService.planJob(jarId, flinkJobRequest);
        } catch(Exception e){
            throw new FlinkWebException("Flink任务提交失败，请检查参数配置是否正确，错误信息", e.getMessage());
        }

        return ResponseResult.success("提交任务成功", flinkWebRestService.runJob(jarId, flinkJobRequest));
    }

    /**
     * 使用SourceAccessId和具体参数执行PostgreSQL CDC任务
     * @param jarId
     * @param flinkJobRequest {@link FlinkJobRequest}<{@link PostgresqlcdcWithSourceAccessIdDtos}>
     * @return {@link ResponseResult}<{@link DwhTableCDCVo}>
     * @throws Exception
     */
    @PostMapping("PostgresqlCdcWithSourceAccessId")
    @Transactional
    @Operation(summary = "提交PostgresqlCDC携带SourceAccessId和具体参数")
    public ResponseResult<DwhTableCDCVo> PostgresqlCdcWithSourceAccessId(@RequestBody FlinkJobRequest<PostgresqlcdcWithSourceAccessIdDtos> flinkJobRequest) throws Exception {
        // TODO: 重新设计,将下列代码封装到service中
        String jarId = flinkJarService.getJarIdbyName(this.cdcJarName);
        Long sourceAccessId = flinkJobRequest.getRequest().getSourceAccessId();
        SourceAccess sourceAccess = sourceAccessService.getById(sourceAccessId);
        if(sourceAccess == null){
            return ResponseResult.fail("sourceAccessId无效，未找到对应的SourceAccess");
        }
        String sourceDatabase = flinkJobRequest.getRequest().getSourceDatabase();
        String sourceTable = flinkJobRequest.getRequest().getSourceTable();

        String targetDatabase = flinkJobRequest.getRequest().getTargetDatabase();
        String targetTable = flinkJobRequest.getRequest().getTargetTable();


        // 创建sourceTable和sourceDatabase
        DataWarehouseDatabase sourceDb = dwhDatabaseService.queryByNameAndSourceAccessIdAndSourceType(sourceDatabase, sourceAccess.getSourceType(), sourceAccessId);
        if (sourceDb == null) {
            sourceDb = dwhDatabaseService.createSourceDatabase(sourceAccessId, sourceDatabase, sourceAccess.getSourceType(), sourceAccess.getDataType());
        }

        DataWarehouseTable sourceTb = dwhTableService.queryByNameAndDatabaseId(null, sourceTable, sourceDb.getId());
        if (sourceTb == null && sourceDb != null) {
            // sourceTable的sourceAccessId为空
            sourceTb = dwhTableService.createTable(sourceDb.getId(), sourceTable, null, -1, "CDC同步的源表");
        }

        // 创建targetTable和targetDatabase
        DataWarehouseDatabase targetDb = dwhDatabaseService.queryByNameAndSourceAccessIdAndSourceType(targetDatabase, "paimon", null);
        if (targetDb == null) {
            if(!this.paimonBaseDao.checkDatabaseExists(targetDatabase)) {
                this.paimonBaseDao.createDatabase(targetDatabase);
            }
            targetDb = dwhDatabaseService.createSourceDatabase(null, targetDatabase, "paimon", 0);
        }

        DataWarehouseTable targetTb = dwhTableService.queryByNameAndDatabaseId(sourceTable, targetTable, targetDb.getId());
        if (targetTb == null && targetDb != null && sourceTb != null) {
            targetTb = dwhTableService.createTable(targetDb.getId(), targetTable, sourceTb.getId(), 0, "ODS层");

        }

        PostgresqlcdcDtos postgresqlcdcDtos = new PostgresqlcdcDtos(
                sourceAccess.getHost(),
                sourceAccess.getPort(),
                sourceAccess.getUsername(),
                sourceAccess.getPassword(),
                sourceDatabase,
                sourceTable,
                targetDatabase,
                targetTable
        );

        flinkJobRequest.setProgramArgsList(postgresqlcdcDtos.getPostgresqlCDCConfigList());
        FlinkJobRequest<PostgresqlcdcDtos> postgresqlRequest = new FlinkJobRequest<>();
        BeanUtils.copyProperties(postgresqlRequest, flinkJobRequest);
        try{
            flinkWebRestService.planJob(jarId, postgresqlRequest);
        } catch(Exception e) {
            throw new FlinkWebException("Flink任务提交失败，请检查参数配置是否正确，错误信息", e.getMessage());
        }
        FlinkJob job = flinkWebRestService.runJob(jarId, postgresqlRequest);
        DwhTableCDCVo sourceTableVo = new DwhTableCDCVo(sourceAccess, sourceDb, sourceTb, targetDb, targetTb, job);
        this.sourceTargetRelationService.save(new TableRelation(sourceAccessId, sourceTable, sourceAccessId, targetTable));

        return ResponseResult.success(sourceTableVo);
    }


    @Operation(summary ="主动雷达CDC")
    @PostMapping("ActiveRadarCDC")
    public ResponseResult<?> activeRadarCDC(@RequestBody FlinkJobRequest<ActiveRadarCDCDto> entity) throws Exception {
        String jarId = flinkJarService.getJarIdbyName(this.radarJarName);
        String database = entity.getRequest().getDatabase();
        String table = entity.getRequest().getTable();
        String description = entity.getRequest().getDescription();
        DataWarehouseDatabase sourceDb = dwhDatabaseService.lambdaQuery()
            .eq(DataWarehouseDatabase::getDatabaseName, database)
            .eq(DataWarehouseDatabase::getSourceType, "paimon")
            .one();
        if(sourceDb == null) {
            throw new CustomException(HttpStatusEnum.DATABASE_NOT_EXISTS);
        }
        DataWarehouseTable sourceTb = dwhTableService.lambdaQuery()
            .eq(DataWarehouseTable::getTableName, table)
            .eq(DataWarehouseTable::getDatabaseId, sourceDb.getId())
            .one();
        if(sourceTb != null) {
            throw new CustomException(HttpStatusEnum.TABLE_ALREADY_EXISTS);
        }
        dwhTableService.createTable(sourceDb.getId(), table, null, 0, description);
        ActiveRadarCDCDto activeRadarCDCDto = new ActiveRadarCDCDto(
            this.endpoint,
            this.accessKey,
            this.secretKey,
            this.warehouse,
            database,
            table,
            entity.getRequest().getMonitor_dir(),
            entity.getRequest().getDuration()
        );
        BeanUtils.copyProperties(entity.getRequest(), activeRadarCDCDto);
        entity.setProgramArgsList(entity.getRequest().getCDCConfigList());
        try {
            flinkWebRestService.planJob(jarId, entity);
            
        } catch (Exception e) {
            throw new FlinkWebException("Flink任务提交失败，请检查参数配置是否正确，错误信息", e.getMessage());
        }
        FlinkJob runJob = flinkWebRestService.runJob(jarId, entity);
        return ResponseResult.success("提交任务成功", runJob);

    }
    @PostMapping("MongoCdc")
    @Operation(summary = "提交MongoDB CDC任务")
    public ResponseResult<FlinkJob> MongoCdc(@RequestBody FlinkJobRequest<MongocdcDtos> flinkJobRequest) throws Exception {
        String jarId = flinkJarService.getJarIdbyName(this.cdcJarName);

        flinkJobRequest.setProgramArgsList(flinkJobRequest.getRequest().getMongoCDCConfigList());

        if(flinkJobRequest.getProgramArgsList() == null){
            throw new FlinkWebException("Flink参数为空,无法提交任务");
        }
        try{
            flinkWebRestService.planJob(jarId, flinkJobRequest);
        } catch(Exception e){
            throw new FlinkWebException("Flink任务提交失败，请检查参数配置是否正确，错误信息", e.getMessage());
        }

        return ResponseResult.success("提交任务成功", flinkWebRestService.runJob(jarId, flinkJobRequest));
    }

    /**
     * 使用SourceAccessId和具体参数执行MongoDB CDC任务
     * @param flinkJobRequest {@link FlinkJobRequest}<{@link MongocdcWithSourceAccessIdDtos}>
     * @return {@link ResponseResult}<{@link DwhTableCDCVo}>
     * @throws Exception
     */
    @PostMapping("MongoCdcWithSourceAccessId")
    @Transactional
    @Operation(summary = "提交MongoDBCDC携带SourceAccessId和具体参数")
    public ResponseResult<DwhTableCDCVo> MongoCdcWithSourceAccessId(@RequestBody FlinkJobRequest<MongocdcWithSourceAccessIdDtos> flinkJobRequest) throws Exception {
        String jarId = flinkJarService.getJarIdbyName(this.cdcJarName);
        Long sourceAccessId = flinkJobRequest.getRequest().getSourceAccessId();
        SourceAccess sourceAccess = sourceAccessService.getById(sourceAccessId);
        if(sourceAccess == null){
            return ResponseResult.fail("sourceAccessId无效，未找到对应的SourceAccess");
        }
        String sourceDatabase = flinkJobRequest.getRequest().getSourceDatabase();
        String sourceTable = flinkJobRequest.getRequest().getSourceTable();
        String targetDatabase = flinkJobRequest.getRequest().getTargetDatabase();
        String targetTable = flinkJobRequest.getRequest().getTargetTable();

        // 创建sourceTable和sourceDatabase
        DataWarehouseDatabase sourceDb = dwhDatabaseService.queryByNameAndSourceAccessIdAndSourceType(sourceDatabase, "mongodb", sourceAccessId);
        if (sourceDb == null) {
            sourceDb = dwhDatabaseService.createSourceDatabase(sourceAccessId, sourceDatabase, "mongodb", 0);
        }
        DataWarehouseTable sourceTb = dwhTableService.queryByNameAndDatabaseId(null, sourceTable, sourceDb.getId());
        if (sourceTb == null && sourceDb != null) {
            // sourceTable的sourceAccessId为空
            sourceTb = dwhTableService.createTable(sourceDb.getId(), sourceTable, null, -1, "CDC同步的源表");
        }

        // 创建targetTable和targetDatabase
        DataWarehouseDatabase targetDb = dwhDatabaseService.queryByNameAndSourceAccessIdAndSourceType(targetDatabase, "paimon", null);
        if (targetDb == null) {
            if(!this.paimonBaseDao.checkDatabaseExists(targetDatabase)) {
                this.paimonBaseDao.createDatabase(targetDatabase);
            }
            targetDb = dwhDatabaseService.createSourceDatabase(null, targetDatabase, "paimon", 0);
        }

        DataWarehouseTable targetTb = dwhTableService.queryByNameAndDatabaseId(sourceTable, targetTable, targetDb.getId());
        if (targetTb == null && targetDb != null && sourceTb != null) {
            targetTb = dwhTableService.createTable(targetDb.getId(), targetTable, sourceTb.getId(), 0, "ODS层");
        }

        MongocdcDtos mongocdcDtos = new MongocdcDtos(
                sourceAccess.getHost() + ":" + sourceAccess.getPort(),
                sourceAccess.getUsername(),
                sourceAccess.getPassword(),
                sourceDatabase,
                sourceTable,
                targetDatabase,
                targetTable
        );

        flinkJobRequest.setProgramArgsList(mongocdcDtos.getMongoCDCConfigList());
        FlinkJobRequest<MongocdcDtos> mongoRequest = new FlinkJobRequest<>();
        BeanUtils.copyProperties(mongoRequest, flinkJobRequest);
        try{
            flinkWebRestService.planJob(jarId, mongoRequest);
        } catch(Exception e) {
            throw new FlinkWebException("Flink任务提交失败，请检查参数配置是否正确，错误信息", e.getMessage());
        }
        FlinkJob job = flinkWebRestService.runJob(jarId, mongoRequest);
        DwhTableCDCVo sourceTableVo = new DwhTableCDCVo(sourceAccess, sourceDb, sourceTb, targetDb, targetTb, job);
        if(sourceTb == null || sourceTb.getId() == null || targetTb == null || targetTb.getId() == null){
            throw new FlinkWebException("SourceTable或TargetTable未创建成功，无法建立关联");
        }else {
            this.sourceTargetRelationService.saveWithSourceTableIdAndTargetTableId(sourceTb.getId(), targetTb.getId());
        }

        return ResponseResult.success(sourceTableVo);
    }

    @PostMapping("MysqlCDC(Batch)")
    public ResponseResult<?> mysqlsync(@RequestBody MysqlcdcWithSourceAccessIdDtos entity) {
        SourceAccess sourceAccess = sourceAccessService.getById(entity.getSourceAccessId());
        if(sourceAccess == null){
            return ResponseResult.fail("sourceAccessId无效，未找到对应的SourceAccess");
        }
        FlinkBaseDao flinkBaseDao = paimonBaseDao.getFlinkBaseDao();
        flinkBaseDao.initFlinkOther(sourceAccess);
        String sourceCatalog = sourceAccess.getConnectionName();
        String sinkCatalog = "paimon_catalog";
        String sourceDatabase = entity.getSourceDatabase();
        String sinkDatabase = entity.getTargetDatabase();
        String sourceTable = entity.getSourceTable();
        String sinkTable = entity.getTargetTable();
        flinkTableDao.copyTable(sourceCatalog, sinkCatalog, sourceDatabase, sinkDatabase, sourceTable, sinkTable);
        return ResponseResult.success("任务提交成功");
    }
    
    
}
